package com.doit.day02;

import com.alibaba.fastjson.JSON;
import lombok.*;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.Properties;

public class _Data2kafka {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        /**
         * 下面的参数是必须要配置的
         */
        //配置kafka集群地址
//        props.setProperty("bootstrap.servers","linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        //key的序列化方式，必须是org.apache.kafka.common.serialization.Serializer接口的实现类
        props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //org.apache.kafka.common.serialization.StringSerializer
//        props.setProperty("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        props.setProperty("value.serializer","org.apache.kafka.common.serialization.StringSerializer");


        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        Person person = new Person();
        ArrayList<String> list = new ArrayList<>();
        list.add("male");
        list.add("female");

        while (true){
            person.setId(RandomUtils.nextInt(100,100000));
            person.setName(RandomStringUtils.randomAlphabetic(10));
            person.setAge(RandomUtils.nextInt(15,99));
            person.setGender(list.get(RandomUtils.nextInt(0,2)));

            //将person对象转换成json串
            String personString = JSON.toJSONString(person);
            ProducerRecord<String, String> record = new ProducerRecord<>("user", personString);
            //发送数据
            producer.send(record);
            //随机休眠
            Thread.sleep(RandomUtils.nextInt(100,300));
        }
    }
}




